﻿using System;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using NetMicro.Cache.RedisProvider;
using NetMicro.Core.Ioc.Attributes;
using NetMicro.Core.Ioc.Dependencies;

namespace NetMicro.Cache.DelayBus
{
    /// <summary>
    /// 延迟消息处理
    /// </summary>
    /// <typeparam name="T"></typeparam>
    public class DelayedMessageProcessor : IDelayedMessageProcessor
    {

        private readonly ILogger<DelayedMessageProcessor> _logger;
        private readonly IRedisCache _redisCache;


        /// <summary>
        /// 延迟消息处理
        /// </summary>
        /// <param name="logger"></param>
        /// <param name="redisCache"></param>
        public DelayedMessageProcessor(
            ILogger<DelayedMessageProcessor> logger,
            IRedisCache redisCache)
        {
            _logger = logger;
            _redisCache = redisCache;
        }
        /// <summary>
        /// 命名service获取
        /// </summary>
        [Dependency]
        public Func<string, ISingleton, object> Func { get; set; }
        /// <summary>
        /// 递送消息到待处理队列
        /// </summary>
        /// <returns></returns>
        public async Task DeliveryToReadyQueueAsync(string topic)
        {
            while (true)
            {
                try
                {
                    using (RedisLock lockKey = _redisCache.Lock($"pub-{topic}", 60))
                    {
                        string jobId = await _redisCache.GetExpireJobsAsync(topic, 1);
                        if (string.IsNullOrWhiteSpace(jobId))
                        {
                            await Task.Delay(500);
                            continue;
                        }
                        //获取任务数据
                        EventJob job = await _redisCache.GetJobAsync<EventJob>(jobId);
                        if (job != null)
                        {
                            using (CSRedis.CSRedisClientPipe<string> tran = _redisCache.StartPipe())
                            {
                                // 设置到待处理任务
                                tran.PushToReadyQueue(topic, job);
                                // 删除当前任务
                                tran.RemoveJob(topic, job.JobId);
                                //添加消息处理任务时间
                                tran.PushJobToBucket(topic, jobId, job.TTR);
                            }
                        }
                        else
                        {
                            await _redisCache.RemoveJobAsync(topic, jobId);
                        }
                    }


                }
                catch (Exception e)
                {
                    _logger.LogError(e, $"DeliveryToReadyQueue Error;Topic:{topic}");
                    await Task.Delay(500);
                }

            }
        }

        /// <summary>
        /// 待处理任务消费处理
        /// </summary>
        /// <returns></returns>
        public async Task ConsumeReadyJobAsync(string topic)
        {


            while (true)
            {
                try
                {
                    using (RedisLock lockKey = _redisCache.Lock($"sub-{topic}", 60))
                    {
                        EventJob job = await _redisCache.GetJobFromReadyQueueAsync<EventJob>(topic);
                        if (job == null)
                        {
                            await Task.Delay(500);
                            continue;
                        }
                        #region 消费任务删除任务
                        await (Func(topic, default) as IDelayProcessor).ExecuteAsync(job);
                        using (CSRedis.CSRedisClientPipe<string> tran = _redisCache.StartPipe())
                        {
                            // 删除任务
                            tran.DelJob(job.JobId);
                            // 删除当前任务
                            tran.RemoveJob(topic, job.JobId);
                        }

                        #endregion
                    }

                }
                catch (Exception e)
                {
                    _logger.LogError(e, $"ConsumeReadyJob Error; Topic:{topic}");
                    await Task.Delay(500);
                }

            }
        }
    }
}
